/// 流式消息传递模块
///
/// 为ProtoActor-RS提供流式消息传递功能，支持大型消息和持续数据流的处理。
/// 可用于实现数据流处理、实时分析和大型消息的分块传输。

use std::any::Any;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use std::future::Future;
use std::pin::Pin;

use async_trait::async_trait;
use tokio::sync::{mpsc, oneshot, Mutex, RwLock};
use thiserror::Error;
use tokio::time::timeout;
use uuid::Uuid;
use tracing::error;

use crate::actor::{Actor, ActorContext, ActorError, DefaultActorContext, PID, Props, SpawnError};
use crate::ActorSystem;

/// 流处理错误
#[derive(Error, Debug)]
pub enum StreamError {
    /// 流已关闭
    #[error("流已关闭")]
    Closed,

    /// 订阅失败
    #[error("订阅失败: {0}")]
    SubscriptionFailed(String),

    /// 推送消息失败
    #[error("消息推送失败: {0}")]
    PushFailed(String),

    /// 处理超时
    #[error("处理超时")]
    Timeout,

    /// 缓冲区已满
    #[error("缓冲区已满")]
    BufferFull,

    /// 未知错误
    #[error("未知错误: {0}")]
    Unknown(String),

    /// 发送错误
    #[error("发送错误")]
    SendError,

    /// 接收错误
    #[error("接收错误")]
    ReceiveError,
}

/// 流命令
#[derive(Debug, Clone)]
pub enum StreamCommand {
    /// 请求更多项目
    Request(usize),
    /// 暂停流
    Pause,
    /// 恢复流
    Resume,
    /// 取消订阅
    Cancel,
    /// 订阅
    Subscribe,
    /// 取消订阅
    Unsubscribe,
}

/// 流状态
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum StreamStatus {
    /// 流动中
    Flowing,
    /// 已暂停
    Paused,
    /// 已关闭
    Closed,
}

/// 流项目
#[derive(Debug, Clone)]
pub struct StreamItem<T: Clone + Send + Sync + 'static> {
    /// 项目内容
    pub value: T,
    /// 项目序号
    pub sequence: u64,
    /// 是否是流的最后一项
    pub is_last: bool,
}

/// 流项目实现
impl<T: Clone + Send + Sync + 'static> StreamItem<T> {
    /// 创建新的流项目
    pub fn new(value: T, sequence: u64) -> Self {
        Self {
            value,
            sequence,
            is_last: false,
        }
    }
}

/// 流发布者接口
#[async_trait]
pub trait StreamPublisher<T: Clone + Send + Sync + 'static>: Send + Sync {
    /// 发布项目到流
    async fn publish(&self, item: T) -> Result<(), StreamError>;
    
    /// 发布多个项目到流
    async fn publish_all(&self, items: Vec<T>) -> Result<(), StreamError>;
    
    /// 完成流
    async fn complete(&self) -> Result<(), StreamError>;
    
    /// 获取订阅者数量
    async fn subscriber_count(&self) -> usize;
}

/// 流订阅者接口 - 参考Go的实现，使用内部可变性
#[async_trait]
pub trait StreamSubscriber<T: Clone + Send + Sync + 'static>: Send + Sync {
    /// 接收下一个项目
    async fn next(&self) -> Result<Option<StreamItem<T>>, StreamError>;
    
    /// 请求更多项目
    async fn request(&self, n: usize) -> Result<(), StreamError>;
    
    /// 暂停流
    async fn pause(&self) -> Result<(), StreamError>;
    
    /// 恢复流
    async fn resume(&self) -> Result<(), StreamError>;
    
    /// 取消订阅
    async fn cancel(&self) -> Result<(), StreamError>;
    
    /// 获取当前状态
    async fn status(&self) -> StreamStatus;
}

/// 流元数据
#[derive(Debug, Clone)]
pub struct StreamMetadata {
    /// 流ID
    pub id: String,
    /// 流主题
    pub topic: String,
    /// 创建时间戳
    pub created_at: i64,
    /// 发布者PID
    pub publisher: Option<PID>,
    /// 自定义元数据
    pub custom: std::collections::HashMap<String, String>,
}

impl StreamMetadata {
    /// 创建新的流元数据
    pub fn new(topic: &str) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            topic: topic.to_string(),
            created_at: chrono::Utc::now().timestamp(),
            publisher: None,
            custom: std::collections::HashMap::new(),
        }
    }
    
    /// 添加自定义元数据
    pub fn with_custom(mut self, key: &str, value: &str) -> Self {
        self.custom.insert(key.to_string(), value.to_string());
        self
    }
    
    /// 设置发布者
    pub fn with_publisher(mut self, publisher: PID) -> Self {
        self.publisher = Some(publisher);
        self
    }
}

/// 流处理器 - 内部实现
struct StreamProcessorImpl<T: Clone + Send + Sync + 'static> {
    /// 流元数据
    metadata: StreamMetadata,
    /// 项目缓冲区
    buffer: VecDeque<StreamItem<T>>,
    /// 订阅者请求计数
    pending_requests: usize,
    /// 流状态
    status: StreamStatus,
    /// 序列号计数器
    sequence: u64,
    /// 最大缓冲区大小
    max_buffer_size: usize,
    /// 订阅者
    subscribers: Vec<mpsc::Sender<StreamItem<T>>>,
    /// 命令通道
    command_tx: mpsc::Sender<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>,
}

impl<T: Clone + Send + Sync + 'static> StreamProcessorImpl<T> {
    /// 创建新的流处理器
    fn new(metadata: StreamMetadata, buffer_size: usize) -> (Self, mpsc::Receiver<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>) {
        let (command_tx, command_rx) = mpsc::channel(32);
        
        let processor = Self {
            metadata,
            buffer: VecDeque::with_capacity(buffer_size),
            pending_requests: 0,
            status: StreamStatus::Flowing,
            sequence: 0,
            max_buffer_size: buffer_size,
            subscribers: Vec::new(),
            command_tx,
        };
        
        (processor, command_rx)
    }
    
    /// 发布项目到流
    async fn publish(&mut self, item: T, is_last: bool) -> Result<(), StreamError> {
        if self.status == StreamStatus::Closed {
            return Err(StreamError::Closed);
        }
        
        // 检查缓冲区是否已满
        if self.buffer.len() >= self.max_buffer_size {
            return Err(StreamError::BufferFull);
        }
        
        // 创建流项目
        let stream_item = StreamItem {
            value: item,
            sequence: self.sequence,
            is_last,
        };
        
        // 增加序列号
        self.sequence += 1;
        
        // 如果有足够的请求，直接发送给订阅者
        if self.pending_requests > 0 && !self.subscribers.is_empty() && self.status == StreamStatus::Flowing {
            self.dispatch_item(stream_item).await;
        } else {
            // 否则，加入缓冲区
            self.buffer.push_back(stream_item);
        }
        
        // 如果是最后一项，关闭流
        if is_last {
            self.status = StreamStatus::Closed;
        }
        
        Ok(())
    }
    
    /// 分发项目到订阅者
    async fn dispatch_item(&mut self, item: StreamItem<T>) {
        let mut failed_subscribers = Vec::new();
        
        // 向每个订阅者发送项目
        for (i, subscriber) in self.subscribers.iter().enumerate() {
            if let Err(_) = subscriber.send(item.clone()).await {
                // 发送失败，订阅者可能已关闭
                failed_subscribers.push(i);
            }
        }
        
        // 移除失败的订阅者（从后向前移除，以避免索引失效）
        for i in failed_subscribers.iter().rev() {
            self.subscribers.remove(*i);
        }
        
        // 减少待处理请求计数
        if self.pending_requests > 0 {
            self.pending_requests -= 1;
        }
    }
    
    /// 处理命令
    async fn handle_command(&mut self, command: &StreamCommand) -> Result<(), StreamError> {
        match command {
            StreamCommand::Request(n) => {
                // 增加待处理请求计数
                self.pending_requests += *n;
                
                // 如果缓冲区有数据且流处于流动状态，尝试分发
                if !self.buffer.is_empty() && self.status == StreamStatus::Flowing {
                    let items_to_send = std::cmp::min(self.buffer.len(), self.pending_requests);
                    
                    for _ in 0..items_to_send {
                        if let Some(item) = self.buffer.pop_front() {
                            self.dispatch_item(item).await;
                        }
                    }
                }
                
                Ok(())
            },
            StreamCommand::Pause => {
                self.status = StreamStatus::Paused;
                Ok(())
            },
            StreamCommand::Resume => {
                self.status = StreamStatus::Flowing;
                
                // 恢复后，尝试分发缓冲区项目
                if !self.buffer.is_empty() && self.pending_requests > 0 {
                    let items_to_send = std::cmp::min(self.buffer.len(), self.pending_requests);
                    
                    for _ in 0..items_to_send {
                        if let Some(item) = self.buffer.pop_front() {
                            self.dispatch_item(item).await;
                        }
                    }
                }
                
                Ok(())
            },
            StreamCommand::Cancel => {
                self.status = StreamStatus::Closed;
                self.buffer.clear();
                self.subscribers.clear();
                Ok(())
            },
            StreamCommand::Subscribe => {
                // 订阅者已经通过 add_subscriber 添加
                Ok(())
            },
            StreamCommand::Unsubscribe => {
                // 取消订阅的处理已经在 dispatch_item 中完成
                Ok(())
            },
        }
    }
    
    /// 添加订阅者
    async fn add_subscriber(&mut self) -> Result<mpsc::Receiver<StreamItem<T>>, StreamError> {
        // 检查流是否已关闭
        if self.status == StreamStatus::Closed {
            return Err(StreamError::Closed);
        }
        
        // 创建订阅者通道
        let (tx, rx) = mpsc::channel(self.max_buffer_size);
        
        // 添加到订阅者列表
        self.subscribers.push(tx);
        
        Ok(rx)
    }
}

/// 流处理器 - 用于发布和管理流
pub struct StreamProcessor<T: Clone + Send + Sync + 'static> {
    /// 流处理器实现
    inner: Arc<Mutex<StreamProcessorImpl<T>>>,
    /// 命令通道
    command_tx: mpsc::Sender<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>,
}

impl<T: Clone + Send + Sync + 'static> StreamProcessor<T> {
    /// 创建新的流处理器
    pub fn new(metadata: StreamMetadata, buffer_size: usize) -> Self {
        let (inner, command_rx) = StreamProcessorImpl::new(metadata, buffer_size);
        let inner = Arc::new(Mutex::new(inner));
        
        // 同步获取命令通道
        let command_tx_clone = {
            let guard = futures::executor::block_on(inner.lock());
            guard.command_tx.clone()
        };
        
        // 启动命令处理循环
        let inner_clone = inner.clone();
        tokio::spawn(async move {
            Self::process_commands(inner_clone, command_rx).await;
        });
        
        Self {
            inner,
            command_tx: command_tx_clone,
        }
    }
    
    /// 处理命令
    async fn process_commands(
        inner: Arc<Mutex<StreamProcessorImpl<T>>>,
        mut command_rx: mpsc::Receiver<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>
    ) {
        while let Some((command, reply_tx)) = command_rx.recv().await {
            let result = {
                let mut processor = inner.lock().await;
                processor.handle_command(&command).await
            };
            
            // 发送结果
            let _ = reply_tx.send(result);
        }
    }
    
    /// 获取流发布者
    pub fn publisher(&self) -> StreamPublisherHandle<T> {
        StreamPublisherHandle {
            inner: self.inner.clone(),
            command_tx: self.command_tx.clone(),
        }
    }
    
    /// 创建新的订阅者
    pub async fn subscribe(&self) -> Result<StreamSubscriberHandle<T>, StreamError> {
        let rx = {
            let mut processor = self.inner.lock().await;
            processor.add_subscriber().await?
        };
        
        Ok(StreamSubscriberHandle {
            rx: Arc::new(Mutex::new(rx)),
            command_tx: self.command_tx.clone(),
            status: Arc::new(RwLock::new(StreamStatus::Flowing)),
        })
    }
    
    /// 获取流元数据
    pub async fn metadata(&self) -> StreamMetadata {
        let processor = self.inner.lock().await;
        processor.metadata.clone()
    }
    
    /// 获取订阅者数量
    pub async fn subscriber_count(&self) -> usize {
        let processor = self.inner.lock().await;
        processor.subscribers.len()
    }
    
    /// 获取当前缓冲区大小
    pub async fn buffer_size(&self) -> usize {
        let processor = self.inner.lock().await;
        processor.buffer.len()
    }
    
    /// 获取当前流状态
    pub async fn status(&self) -> StreamStatus {
        let processor = self.inner.lock().await;
        processor.status
    }
}

/// 流发布者句柄
pub struct StreamPublisherHandle<T: Clone + Send + Sync + 'static> {
    /// 流处理器
    inner: Arc<Mutex<StreamProcessorImpl<T>>>,
    /// 命令通道
    command_tx: mpsc::Sender<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>,
}

#[async_trait]
impl<T: Clone + Send + Sync + 'static> StreamPublisher<T> for StreamPublisherHandle<T> {
    /// 发布项目到流
    async fn publish(&self, item: T) -> Result<(), StreamError> {
        let mut processor = self.inner.lock().await;
        processor.publish(item, false).await
    }
    
    /// 发布多个项目到流
    async fn publish_all(&self, items: Vec<T>) -> Result<(), StreamError> {
        let mut processor = self.inner.lock().await;
        
        let items_len = items.len();
        for (i, item) in items.into_iter().enumerate() {
            let is_last = i == items_len - 1;
            processor.publish(item, is_last).await?;
        }
        
        Ok(())
    }
    
    /// 完成流
    async fn complete(&self) -> Result<(), StreamError> {
        // 发送一个标记为最后的空消息
        let (tx, rx) = oneshot::channel();
        self.command_tx.send((StreamCommand::Cancel, tx)).await
            .map_err(|_| StreamError::Closed)?;
        
        rx.await
            .map_err(|_| StreamError::Closed)?
    }
    
    /// 获取订阅者数量
    async fn subscriber_count(&self) -> usize {
        let processor = self.inner.lock().await;
        processor.subscribers.len()
    }
}

impl<T: Clone + Send + Sync + 'static> Clone for StreamPublisherHandle<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            command_tx: self.command_tx.clone(),
        }
    }
}

/// 流订阅者句柄
pub struct StreamSubscriberHandle<T: Clone + Send + Sync + 'static> {
    /// 接收通道
    rx: Arc<Mutex<mpsc::Receiver<StreamItem<T>>>>,
    /// 命令通道
    command_tx: mpsc::Sender<(StreamCommand, oneshot::Sender<Result<(), StreamError>>)>,
    /// 本地状态
    status: Arc<RwLock<StreamStatus>>,
}

// 为 StreamSubscriberHandle 实现 Clone
impl<T: Clone + Send + Sync + 'static> Clone for StreamSubscriberHandle<T> {
    fn clone(&self) -> Self {
        // 接收器无法被克隆，因此我们仅克隆它的包装器
        Self {
            rx: self.rx.clone(),
            command_tx: self.command_tx.clone(),
            status: self.status.clone(),
        }
    }
}

#[async_trait]
impl<T: Clone + Send + Sync + 'static> StreamSubscriber<T> for StreamSubscriberHandle<T> {
    /// 接收下一个项目
    async fn next(&self) -> Result<Option<StreamItem<T>>, StreamError> {
        // 检查状态
        let status = *self.status.read().await;
        if status == StreamStatus::Closed {
            return Err(StreamError::Closed);
        }
        
        // 获取锁并使用通道
        let mut rx_guard = self.rx.lock().await;
        match timeout(Duration::from_secs(1), rx_guard.recv()).await {
            Ok(Some(item)) => {
                // 如果是最后一项，更新状态
                if item.is_last {
                    let mut status = self.status.write().await;
                    *status = StreamStatus::Closed;
                }
                Ok(Some(item))
            },
            Ok(None) => {
                // 流已关闭
                let mut status = self.status.write().await;
                *status = StreamStatus::Closed;
                Ok(None)
            },
            Err(_) => {
                // 超时
                Err(StreamError::Timeout)
            }
        }
    }
    
    /// 请求更多项目
    async fn request(&self, n: usize) -> Result<(), StreamError> {
        let (tx, rx) = oneshot::channel();
        self.command_tx.send((StreamCommand::Request(n), tx)).await
            .map_err(|_| StreamError::Closed)?;
        
        rx.await
            .map_err(|_| StreamError::Closed)?
    }
    
    /// 暂停流
    async fn pause(&self) -> Result<(), StreamError> {
        // 更新本地状态
        let mut status = self.status.write().await;
        *status = StreamStatus::Paused;
        
        // 发送暂停命令
        let (tx, rx) = oneshot::channel();
        self.command_tx.send((StreamCommand::Pause, tx)).await
            .map_err(|_| StreamError::Closed)?;
        
        rx.await
            .map_err(|_| StreamError::Closed)?
    }
    
    /// 恢复流
    async fn resume(&self) -> Result<(), StreamError> {
        // 更新本地状态
        let mut status = self.status.write().await;
        *status = StreamStatus::Flowing;
        
        // 发送恢复命令
        let (tx, rx) = oneshot::channel();
        self.command_tx.send((StreamCommand::Resume, tx)).await
            .map_err(|_| StreamError::Closed)?;
        
        rx.await
            .map_err(|_| StreamError::Closed)?
    }
    
    /// 取消订阅
    async fn cancel(&self) -> Result<(), StreamError> {
        // 更新本地状态
        let mut status = self.status.write().await;
        *status = StreamStatus::Closed;
        
        // 发送取消命令
        let (tx, rx) = oneshot::channel();
        self.command_tx.send((StreamCommand::Cancel, tx)).await
            .map_err(|_| StreamError::Closed)?;
        
        rx.await
            .map_err(|_| StreamError::Closed)?
    }
    
    /// 获取当前状态
    async fn status(&self) -> StreamStatus {
        *self.status.read().await
    }
}

/// 流消息 - Actor消息类型
#[derive(Debug, Clone)]
pub enum StreamMessage<T: Clone + Send + Sync + 'static> {
    /// 新的流数据
    Data(StreamItem<T>),
    /// 流命令
    Command(StreamCommand),
    /// 流已完成
    Complete,
    /// 流错误
    Error(String),
}

/// 流Actor - 基于Actor的流处理
pub struct StreamActor<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> {
    /// 流处理器
    processor: StreamProcessor<T>,
    /// 流处理器
    handler: H,
    /// 订阅者句柄
    subscriber: Option<StreamSubscriberHandle<T>>,
    /// 批处理大小
    batch_size: usize,
    /// 处理中的项目
    processing: Vec<StreamItem<T>>,
}

/// 流处理器接口 - 处理流数据
#[async_trait]
pub trait StreamHandler<T: Clone + Send + Sync + 'static>: Send + Sync + Clone + 'static {
    /// 处理流项目 - 添加 Send 约束
    async fn handle(&mut self, ctx: &mut impl ActorContext, item: StreamItem<T>) -> Result<(), ActorError>;
    
    /// 流已完成 - 添加 Send 约束
    async fn on_complete(&mut self, ctx: &mut impl ActorContext) -> Result<(), ActorError>;
    
    /// 流错误 - 添加 Send 约束
    async fn on_error(&mut self, ctx: &mut impl ActorContext, error: String) -> Result<(), ActorError>;
}

impl<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> StreamActor<T, H> {
    /// 创建新的流Actor
    pub fn new(handler: H, batch_size: usize) -> Self {
        let metadata = StreamMetadata::new("default");
        let processor = StreamProcessor::new(metadata, 100);
        
        Self {
            processor,
            handler,
            subscriber: None,
            batch_size,
            processing: Vec::new(),
        }
    }
    
    /// 从订阅者创建流Actor
    pub fn from_subscriber(handler: H, subscriber: StreamSubscriberHandle<T>, batch_size: usize) -> Self {
        let metadata = StreamMetadata::new("default");
        let processor = StreamProcessor::new(metadata, 100);
        
        Self {
            processor,
            handler,
            subscriber: Some(subscriber),
            batch_size,
            processing: Vec::new(),
        }
    }
    
    /// 从处理器创建流Actor
    pub fn from_processor(handler: H, processor: StreamProcessor<T>, batch_size: usize) -> Self {
        Self {
            processor,
            handler,
            subscriber: None,
            batch_size,
            processing: Vec::new(),
        }
    }
    
    /// 请求更多项目
    async fn request_more(&self) -> Result<(), StreamError> {
        if let Some(ref subscriber) = self.subscriber {
            subscriber.request(self.batch_size).await?;
        }
        Ok(())
    }
}

#[async_trait]
impl<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> Actor for StreamActor<T, H> {
    type Context = DefaultActorContext;
    
    async fn started(&self, ctx: &mut Self::Context) -> Result<(), ActorError> {
        if let Some(ref subscriber) = self.subscriber {
            subscriber.request(self.batch_size).await
                .map_err(|e| ActorError::Unknown(e.to_string()))?;
        }
        Ok(())
    }
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        // 处理流消息
        if let Some(stream_msg) = msg.downcast_ref::<StreamMessage<T>>() {
            // 克隆处理器而不是使用 &self 引用
            let mut handler = self.handler.clone();
            
            match stream_msg {
                StreamMessage::Data(item) => {
                    // 克隆项目
                    let item_clone = item.clone();
                    
                    // 处理数据项
                    handler.handle(ctx, item_clone).await?;
                    
                    // 如果是最后一项，通知完成
                    if item.is_last {
                        handler.on_complete(ctx).await?;
                    } else {
                        // 否则，如果有订阅者，请求更多数据
                        if let Some(ref subscriber) = self.subscriber {
                            subscriber.request(1).await
                                .map_err(|e| ActorError::Unknown(e.to_string()))?;
                        }
                    }
                },
                StreamMessage::Command(cmd) => {
                    // 处理命令
                    match cmd {
                        StreamCommand::Request(n) => {
                            // 请求更多数据
                            if let Some(ref subscriber) = self.subscriber {
                                subscriber.request(*n).await
                                    .map_err(|e| ActorError::Unknown(e.to_string()))?;
                            }
                        },
                        StreamCommand::Pause => {
                            // 暂停流
                            if let Some(ref subscriber) = self.subscriber {
                                subscriber.pause().await
                                    .map_err(|e| ActorError::Unknown(e.to_string()))?;
                            }
                        },
                        StreamCommand::Resume => {
                            // 恢复流
                            if let Some(ref subscriber) = self.subscriber {
                                subscriber.resume().await
                                    .map_err(|e| ActorError::Unknown(e.to_string()))?;
                            }
                        },
                        StreamCommand::Cancel => {
                            // 取消订阅
                            if let Some(ref subscriber) = self.subscriber {
                                subscriber.cancel().await
                                    .map_err(|e| ActorError::Unknown(e.to_string()))?;
                            }
                            
                            // 通知完成
                            handler.on_complete(ctx).await?;
                        },
                        StreamCommand::Subscribe => {
                            // 订阅者已经通过 add_subscriber 添加
                        },
                        StreamCommand::Unsubscribe => {
                            // 取消订阅的处理已经在 dispatch_item 中完成
                        },
                    }
                },
                StreamMessage::Complete => {
                    // 流已完成
                    handler.on_complete(ctx).await?;
                },
                StreamMessage::Error(error) => {
                    // 流错误
                    handler.on_error(ctx, error.clone()).await?;
                },
            }
        }
        
        Ok(())
    }
}

/// 为 StreamProcessor 创建流Actor
pub async fn create_stream_actor<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static>(
    system: ActorSystem,
    stream_id: &str,
    handler: H,
    batch_size: usize,
) -> Result<PID, ActorError> {
    let factory = StreamFactory::new(system);
    factory.create_stream_actor(stream_id, handler, batch_size).await
}

/// 流工厂
pub struct StreamFactory {
    /// Actor系统
    system: ActorSystem,
    /// 流注册表
    streams: Arc<RwLock<std::collections::HashMap<String, Arc<dyn Any + Send + Sync>>>>,
}

impl StreamFactory {
    /// 创建新的流工厂
    pub fn new(system: ActorSystem) -> Self {
        StreamFactory {
            system,
            streams: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 创建新的流处理器
    pub async fn create_stream<T: Clone + Send + Sync + 'static>(&self, topic: &str, buffer_size: usize) -> StreamProcessor<T> {
        let metadata = StreamMetadata::new(topic);
        let stream_id = metadata.id.clone();
        let processor = StreamProcessor::new(metadata, buffer_size);
        
        // 存储流处理器到注册表
        let streams = self.streams.clone();
        let mut streams_write = streams.write().await;
        streams_write.insert(stream_id, Arc::new(processor.clone()));
        processor
    }

    /// 获取指定ID的流处理器
    pub async fn get_stream<T: Clone + Send + Sync + 'static>(&self, stream_id: &str) -> Option<StreamProcessor<T>> {
        let streams = self.streams.read().await;
        streams.get(stream_id)
            .and_then(|s| s.clone().downcast_ref::<StreamProcessor<T>>().cloned())
    }

    /// 获取所有流ID
    pub async fn get_all_stream_ids(&self) -> Vec<String> {
        let streams = self.streams.read().await;
        streams.keys().cloned().collect()
    }

    /// 移除流
    pub async fn remove_stream(&self, stream_id: &str) -> bool {
        let mut streams = self.streams.write().await;
        streams.remove(stream_id).is_some()
    }

    /// 创建流Actor
    pub async fn create_stream_actor<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static>(
        &self,
        stream_id: &str,
        handler: H,
        batch_size: usize,
    ) -> Result<PID, ActorError> {
        // 查找流
        let stream = self.get_stream::<T>(stream_id).await
            .ok_or_else(|| ActorError::Unknown(format!("Stream not found: {}", stream_id)))?;
        
        // 订阅流
        let subscriber = stream.subscribe().await
            .map_err(|e| ActorError::Unknown(e.to_string()))?;
        
        // 创建Actor
        let actor = StreamActor::from_subscriber(handler, subscriber, batch_size);
        
        // 使用Props::new创建actor实例
        let actor_clone = actor.clone();
        let props = Props::new(move || Box::new(actor_clone.clone()));
        
        // 转换错误类型
        self.system.spawn::<StreamActor<T, H>>(props).await.map_err(ActorError::from)
    }
}

impl From<SpawnError> for ActorError {
    fn from(err: SpawnError) -> Self {
        ActorError::Unknown(err.to_string())
    }
}

// 为 StreamProcessor 实现 Clone
impl<T: Clone + Send + Sync + 'static> Clone for StreamProcessor<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            command_tx: self.command_tx.clone(),
        }
    }
}

// 为 StreamActor 实现 Clone
impl<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> Clone for StreamActor<T, H> {
    fn clone(&self) -> Self {
        Self {
            processor: self.processor.clone(),
            handler: self.handler.clone(),
            subscriber: self.subscriber.clone(),
            batch_size: self.batch_size,
            processing: Vec::new(),
        }
    }
}

// 为 StreamActor 实现 Send + Sync
unsafe impl<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> Send for StreamActor<T, H> {}
unsafe impl<T: Clone + Send + Sync + 'static, H: StreamHandler<T> + Send + Sync + 'static> Sync for StreamActor<T, H> {}

// 添加 StreamError -> ActorError 的转换
impl From<StreamError> for ActorError {
    fn from(err: StreamError) -> Self {
        match err {
            StreamError::Closed => ActorError::ActorFailure("Stream closed".to_string()),
            StreamError::Timeout => ActorError::Timeout,
            _ => ActorError::Unknown(err.to_string()),
        }
    }
} 